package com.boilermaker.mq.mqserver;

import com.boilermaker.mq.common.*;
import com.boilermaker.mq.common.request_payload.*;
import com.boilermaker.mq.common.response_payload.BasicReturns;
import com.boilermaker.mq.common.response_payload.SubscribeReturns;
import com.boilermaker.mq.mqserver.core.BasicProperties;
import lombok.extern.slf4j.Slf4j;

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Slf4j
public class BrokerServer {
    private ServerSocket serverSocket;
    // TODO: 可实现支持多个虚拟主机，需要添加针对虚拟主机的增删查改方法
    private VirtualHost virtualHost = new VirtualHost("default");
    // 记录当前与服务器通信的客户端，以及每个客户端内部的 Channel。结构：key-channelId val-Socket
    private ConcurrentHashMap<String, Socket> sessions = new ConcurrentHashMap<>();
    // 处理多个客户端的请求
    private ExecutorService executorService;
    // 控制服务器是否继续运行。volatile 关键字保证 runnable 的修改能够被 while 循环感知到。
    private volatile boolean runnable = true;

    public BrokerServer(int port) throws IOException {
        serverSocket = new ServerSocket(port);
    }

    public void start() throws IOException {
        log.info("启动服务器");
        executorService = Executors.newCachedThreadPool();
        try {
            while (runnable) {
                // 服务器将持续运行，直至某线程调用 stop，此时服务器线程将触发 SocketException
                Socket clientSocket = serverSocket.accept();
                executorService.submit(() -> processConnection(clientSocket));
            }
        } catch (SocketException e) {
            log.info("服务器终止");
        }
    }

    public void stop() throws IOException {
        runnable = false;
        executorService.shutdownNow();
        serverSocket.close();
    }

    private void processConnection(Socket clientSocket) {
        // 先获取流对象，再读取格式化数据
        try (DataOutputStream dataOutputStream = new DataOutputStream(clientSocket.getOutputStream());
             DataInputStream dataInputStream = new DataInputStream(clientSocket.getInputStream())) {
            while (true) {
                // 读取请求并解析
                Request request = readRequest(dataInputStream);
                // 根据请求计算响应
                Response response = process(request, clientSocket);
                // 返回响应给客户端
                writeResponse(response, dataOutputStream);
            }
        } catch (EOFException | SocketException e) { // DataInputStream 读到文件末尾或客户端终止程序，视为连接正常关闭
            log.info("连接关闭, 客户端地址: " + clientSocket.getInetAddress().toString() + ":" + clientSocket.getPort());
        } catch (IOException | ClassNotFoundException | MqException e) {
            log.info("连接异常");
            e.printStackTrace();
        } finally {
            try {
                clientSocket.close(); // 关闭 Socket
                clearClosedSession(clientSocket); // 清理 Socket 内部的 Channel
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    // 读取请求并解析，请求格式：type(4Bytes) length(4Bytes) payload(length)
    private Request readRequest(DataInputStream dataInputStream) throws IOException {
        Request request = new Request();
        request.setType(dataInputStream.readInt());
        request.setLength(dataInputStream.readInt());

        byte[] payload = new byte[request.getLength()];
        int actualLength = dataInputStream.read(payload);
        if (actualLength != payload.length) throw new IOException("格式有误");
        request.setPayload(payload);

        return request;
    }

    // 返回响应给客户端，响应格式：type(4Bytes) length(4Bytes) payload(length)
    private void writeResponse(Response response, DataOutputStream dataOutputStream) throws IOException {
        dataOutputStream.writeInt(response.getType());
        dataOutputStream.writeInt(response.getLength());
        dataOutputStream.write(response.getPayload());

        dataOutputStream.flush();
    }

    private Response process(Request request, Socket clientSocket) throws IOException, ClassNotFoundException, MqException {
        // 解析请求载荷
        BasicArguments basicArguments = (BasicArguments) BinaryTool.fromBytes(request.getPayload());
        log.info("解析请求: rid=" + basicArguments.getRid() + ", channelId=" + basicArguments.getChannelId() +
                ", type=" + request.getType() + ", length=" + request.getLength());
        // 解析请求类型，根据类型将载荷向下转型
        boolean ok = true;
        switch (request.getType()) {
            case 0x1: { // 创建 Channel
                sessions.put(basicArguments.getChannelId(), clientSocket);
                log.info("已创建 Channel, channelId=" + basicArguments.getChannelId());
                break;
            }
            case 0x2: { // 销毁 Channel
                sessions.remove(basicArguments.getChannelId());
                break;
            }
            case 0x3: { // 创建 Exchange
                ExchangeDeclareArguments arguments = (ExchangeDeclareArguments) basicArguments;
                ok = virtualHost.exchangeDeclare(arguments.getExchangeName(), arguments.getExchangeType(),
                        arguments.isDurable(), arguments.isAutoDelete(), arguments.getArguments());
                break;
            }
            case 0x4: { // 删除 Exchange
                ExchangeDeleteArguments arguments = (ExchangeDeleteArguments) basicArguments;
                ok = virtualHost.exchangeDelete(arguments.getExchangeName());
                break;
            }
            case 0x5: { // 创建 Queue
                QueueDeclareArguments arguments = (QueueDeclareArguments) basicArguments;
                ok = virtualHost.queueDeclare(arguments.getQueueName(), arguments.isDurable(),
                        arguments.isExclusive(), arguments.isAutoDelete(), arguments.getArguments());
                break;
            }
            case 0x6: { // 删除 Queue
                QueueDeleteArguments arguments = (QueueDeleteArguments) basicArguments;
                ok = virtualHost.queueDelete(arguments.getQueueName());
                break;
            }
            case 0x7: { // 创建 Binding
                QueueBindArguments arguments = (QueueBindArguments) basicArguments;
                ok = virtualHost.queueBind(arguments.getExchangeName(), arguments.getQueueName(), arguments.getBindingKey());
                break;
            }
            case 0x8: { // 删除 Binding
                QueueUnbindArguments arguments = (QueueUnbindArguments) basicArguments;
                ok = virtualHost.queueUnbind(arguments.getExchangeName(), arguments.getQueueName());
                break;
            }
            case 0x9: { // 发布 Message
                BasicPublishArguments arguments = (BasicPublishArguments) basicArguments;
                ok = virtualHost.basicPublish(arguments.getExchangeName(), arguments.getRoutingKey(),
                        arguments.getBasicProperties(), arguments.getBody());
                break;
            }
            case 0xa: { // 消费 Message
                BasicConsumeArguments arguments = (BasicConsumeArguments) basicArguments;
                ok = virtualHost.basicConsume(arguments.getConsumeTag(), arguments.getQueueName(), arguments.isAutoAck(),
                        (String consumerTag, BasicProperties basicProperties, byte[] body) -> {
                            // 该回调函数将收到的消息推送至客户端（执行于每次推送消息）
                            Socket socket = sessions.get(consumerTag);
                            if (socket == null || socket.isClosed()) {
                                throw new MqException("无法推送，因为订阅消息的客户端已经关闭");
                            }
                            // 构造响应数据
                            SubscribeReturns subscribeReturns = new SubscribeReturns();
                            subscribeReturns.setChannelId(consumerTag);
                            subscribeReturns.setRid("");
                            subscribeReturns.setOk(true);
                            subscribeReturns.setConsumerTag(consumerTag);
                            subscribeReturns.setBasicProperties(basicProperties);
                            subscribeReturns.setBody(body);

                            byte[] payload = BinaryTool.toBytes(subscribeReturns);

                            Response response = new Response();
                            response.setType(0xc); // 0xc 表示该响应类型是消息数据
                            response.setLength(payload.length);
                            response.setPayload(payload);

                            DataOutputStream dataOutputStream = new DataOutputStream(clientSocket.getOutputStream());
                            writeResponse(response, dataOutputStream);
                        });
                break;
            }
            case 0xb: { // 手动确认应答
                BasicAckArguments arguments = (BasicAckArguments) basicArguments;
                ok = virtualHost.basicAck(arguments.getQueueName(), arguments.getMessageId());
                break;
            }
            default: {
                throw new MqException("未知请求类型, type=" + request.getType());
            }
        }
        // 构造响应
        BasicReturns basicReturns = new BasicReturns();
        basicReturns.setChannelId(basicArguments.getChannelId());
        basicReturns.setRid(basicArguments.getRid());
        basicReturns.setOk(ok);

        byte[] payload = BinaryTool.toBytes(basicReturns);

        Response response = new Response();
        response.setLength(payload.length);
        response.setType(request.getType()); // 响应类型与请求相匹配
        response.setPayload(payload);
        log.info("发送响应: rid=" + basicReturns.getRid() + ", channelId=" + basicReturns.getChannelId() +
                ", type=" + response.getType() + ", length=" + response.getLength());

        return response;
    }

    // 该方法于客户端与服务器断开连接时被调用
    // 此时 sessions 中可能尚有未正常删除的 channelId
    private void clearClosedSession(Socket clientSocket) {
        // 为避免在遍历的同时删除，导致迭代器失效，此处需要将要删除的 channelId 存起来
        List<String> toDelete = new ArrayList<>();
        // 仅查找的循环
        for (Map.Entry<String, Socket> entry : sessions.entrySet()) {
            if (entry.getValue() == clientSocket) toDelete.add(entry.getKey());
        }
        // 仅删除的循环
        for (String id : toDelete) {
            sessions.remove(id);
        }
        log.info("已清理残余会话, 删除的会话为: " + toDelete);
    }
}
